package org.leo.dm.example.writer;

import java.sql.Connection;
import java.util.HashMap;
import java.util.Map;

import org.leo.dm.common.bean.Constants;
import org.leo.dm.common.bean.StepResultBean;
import org.leo.dm.common.exception.DpException;
import org.leo.dm.common.exception.DpExceptionCodeEnum;
import org.leo.dm.component.callback.DbExceptionCallback;
import org.leo.dm.component.callback.DefaultDbExceptionCallback;
import org.leo.dm.component.util.DbBatch;
import org.leo.dm.component.util.DbBean;
import org.leo.dm.component.util.DbManager;
import org.leo.dm.component.util.DbWriteHelper;
import org.leo.dm.component.util.DmlEnum;
import org.leo.dm.component.writer.Writer;
import org.leo.dm.util.ObjectUtil;
import org.leo.dm.util.ToUtil;

/**
 * 
 * Created on 2016年2月19日
 * <p>
 * Description db写组件实现类
 * <p>
 * Copyright neusoft (c) 2015 .
 * 
 * @author xiao_h
 *
 */
public class DbWriter extends Writer {

	public DbWriter() {
		super();
		// 初始化相关插件
		this.dbExceptionCallback = new DefaultDbExceptionCallback();
	}

	protected DmlEnum dmlEnum;

	public void setDmlEnum(DmlEnum dmlEnum) {
		this.dmlEnum = dmlEnum;
	}

	private DbWriteHelper dbWriteHelper = new DbWriteHelper();

	public DbWriteHelper getDbWriteHelper() {
		return dbWriteHelper;
	}

	protected int updateWhereSplitIndex;

	protected int batchSize;

	public void setBatchSize(int batchSize) {
		this.batchSize = batchSize;
	}

	/**
	 * 
	 * Created on 2016年1月21日 .
	 * 
	 * @author xiao_h
	 * @param updateWhereSplitIndex
	 *            从0开始
	 */
	public void setUpdateWhereSplitIndex(int updateWhereSplitIndex) {
		this.updateWhereSplitIndex = updateWhereSplitIndex;
	}

	private Map<String, DbBatch> dbBatchHolder = new HashMap<String, DbBatch>();

	private Map<String, DbBean> tableDbInfo;

	public void setTableDbInfo(Map<String, DbBean> tableDbInfo) {
		this.tableDbInfo = tableDbInfo;
	}

	protected Connection conn;

	protected Map<String, String> tableNameMapping;

	public void setTableNameMapping(Map<String, String> tableNameMapping) {
		this.tableNameMapping = tableNameMapping;
	}

	protected Map<String, Map<String, String>> tableColumnMapping;

	public void setTableColumnMapping(Map<String, Map<String, String>> tableColumnMapping) {
		this.tableColumnMapping = tableColumnMapping;
	}

	private DbExceptionCallback dbExceptionCallback;

	public void setDbExceptionCallback(DbExceptionCallback dbExceptionCallback) {
		this.dbExceptionCallback = dbExceptionCallback;
	}

	protected String[] targetTableNames;

	public void setConn(Connection conn) {
		this.conn = conn;
	}

	@Override
	public StepResultBean call() {
		DbBean dbBean = null;
		StepResultBean resultBean = new StepResultBean();

		DbBatch batchUtil = new DbBatch();
		dbWriteHelper.setAlreadyRunCount(0);

		try {
			if (dmlEnum == null) {
				throw new DpException(DpExceptionCodeEnum.PARAM_ERROR, "DML类型为空 无法执行");
			}

			Map<Object, Object> mapTemp = null;

			Object[] targetSetDataNames = null;
			Object[] targetWhereDataNames = null;

			String targetTableName = null;
			Map<String, String> columnMapping = null;
			Object[] columnNames = null;
			Object[] data = null;
			Connection connTemp = null;
			while (isRunFlag()) {
				data = storage.getData();
				counter++;// 计数器累加

				// 非业务数据
				if (Constants.NONBUSINESSDATA.equals(data[0]) && Constants.QUEUE.equals(data[1]) && Constants.START.equals(data[2])) {
					ToUtil.to("开始接收数据");
					continue;
				} else if (Constants.NONBUSINESSDATA.equals(data[0]) && Constants.SEGMENT.equals(data[1]) && Constants.START.equals(data[2])) {
					// 该判断内要做的是 初始化一次新的插入表操作所需的内容
					// 取出元数据对应的columnname
					columnNames = (Object[]) data[5];
					// 取出元数据对应的targetTableName

					if (tableNameMapping != null && tableNameMapping.containsKey(data[4])) {
						targetTableName = tableNameMapping.get(data[4]);
					} else {
						targetTableName = (String) data[4];
					}
					// 初始化字段映射关系
					columnMapping = null;
					// if 存在映射
					if (tableColumnMapping != null && tableColumnMapping.containsKey(targetTableName)) {
						columnMapping = tableColumnMapping.get(targetTableName);
						targetDataNames = columnMapping.keySet().toArray(new Object[0]);
					} else {
						targetDataNames = (Object[]) data[5];
					}

					if (dbBatchHolder.containsKey(targetTableName)) {
						batchUtil = dbBatchHolder.get(targetTableName);
					} else {

						if (tableDbInfo != null && tableDbInfo.containsKey(targetTableName)) {
							dbBean = tableDbInfo.get(targetTableName);
							connTemp = DbManager.getConection(dbBean);

							this.conn = connTemp;
						}

						switch (dmlEnum) {
						case INSERT:
							batchUtil = new DbBatch(conn, targetTableName, DmlEnum.INSERT, targetDataNames, new Object[] {});
							break;
						case UPDATE:
							if (updateWhereSplitIndex <= 0 || updateWhereSplitIndex >= targetDataNames.length) {
								throw new DpException(DpExceptionCodeEnum.PARAM_ERROR, "更新操作时 设置更新条件分隔符 值错误，不能为 <=0 或 >=max");
							}
							targetSetDataNames = new Object[updateWhereSplitIndex];
							targetWhereDataNames = new Object[targetDataNames.length - updateWhereSplitIndex];
							System.arraycopy(targetDataNames, 0, targetSetDataNames, 0, updateWhereSplitIndex);
							System.arraycopy(targetDataNames, updateWhereSplitIndex, targetWhereDataNames, 0, targetDataNames.length - updateWhereSplitIndex);
							batchUtil = new DbBatch(conn, targetTableName, DmlEnum.UPDATE, targetSetDataNames, targetWhereDataNames);
							break;
						case DELETE:
							batchUtil = new DbBatch(conn, targetTableName, DmlEnum.DELETE, new Object[] {}, targetDataNames);
							break;
						}

						if (batchSize != 0) {
							batchUtil.setBatchSize(batchSize);
						}
						batchUtil.setBatchHelper(dbWriteHelper);
						batchUtil.setDbExceptionCallback(dbExceptionCallback);
						dbBatchHolder.put(targetTableName, batchUtil);
					}

					continue;
				} else if (Constants.NONBUSINESSDATA.equals(data[0]) && Constants.SEGMENT.equals(data[1]) && Constants.END.equals(data[2])) {
					continue;
				} else if (Constants.NONBUSINESSDATA.equals(data[0]) && Constants.QUEUE.equals(data[1]) && Constants.END.equals(data[2])) {
					for (Map.Entry<String, DbBatch> entry : dbBatchHolder.entrySet()) {
						ToUtil.to("数据写入执行完成");
						entry.getValue().last();
					}
					setRunFlag(false);
					Map<String, Object> callBackMap = new HashMap<String, Object>();
					callBackMap.put("counter", counter);
					callback.handle(callBackMap);
					continue;
				} else if (Constants.NONBUSINESSDATA.equals(data[0]) && Constants.ERROR.equals(data[1])) {
					setRunFlag(false);
					continue;
				}

				// 判断 是否配置了字段映射
				if (columnMapping != null) {
					mapTemp = ObjectUtil.arrayToMap(columnNames, data);
					targetDataValues = new String[targetDataNames.length];
					for (int i = 0; i < targetDataNames.length; i++) {
						targetDataValues[i] = mapTemp.get(columnMapping.get(targetDataNames[i]));
					}
				} else {
					targetDataValues = (Object[]) data;
				}

				batchUtil.handle(targetDataValues);

			}
			ToUtil.to("退出数据写入执行");
			resultBean.setCode("1");
			resultBean.setName("运行正常完成");
			resultBean.setSuccessCount(dbWriteHelper.getAlreadyRunCount());
		} catch (Exception e) {
			e.printStackTrace();
			resultBean.setCode("-1");
			resultBean.setName("运行发生异常");
			resultBean.setErrorDesc(e.getMessage());
			try {
				this.getFlow().stop();
			} catch (Exception e1) {
				e1.printStackTrace();
			}
		} finally {
			for (Map.Entry<String, DbBatch> entry : dbBatchHolder.entrySet()) {
				if (entry.getValue() != null) {
					try {
						entry.getValue().close();
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
			counter = 0;// 计数器清零
		}

		return resultBean;
	}

}
